package com.naiterui.ehp.bs.agent.websocket;

import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import com.naiterui.common.redis.RedisUtil;
import com.naiterui.ehp.bs.agent.util.RedisKeyUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * websocket协议路径处理、连接鉴权处理
 *
 * @author gaoliangliang
 * @since 2022/2/22
 */
@ServerEndpoint(value = "/websocket/{userId}/{token}")
@Component
@Slf4j
public class WebsocketSession {

    private static final ConcurrentHashMap<String, WebsocketSession> websocketSessionMap = new ConcurrentHashMap<>();
    private Session session;
    private String userId;
    @SuppressWarnings("all")
    // TODO token保留，若扩展多设备登录可用与区分多设备token
    private String token;

    @SneakyThrows
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId, @PathParam("token") String token) {
        this.session = session;
        this.userId = StrUtil.trim(userId);
        this.token = StrUtil.trim(token);

        // 处理用户身份鉴权
        if (StrUtil.hasBlank(userId, token) || !StrUtil.equals(token, RedisUtil.valueOps().getString(RedisKeyUtil.getLoginTokenKey(Long.valueOf(userId))))) {
            this.session.close(new CloseReason(() -> 9999, "连接校验失败:logout"));
            return;
        }
        WebsocketSession websocketSession = websocketSessionMap.putIfAbsent(userId, this);

        // 目前限制单设备登录
        if (websocketSession != null && websocketSession.session != null) {
            websocketSession.session.close(new CloseReason(() -> 9999, "连接校验失败:logout"));
        }

        log.info("【用户连接：{} {}】,【目前总数：{}】", this.userId, this.token, websocketSessionMap.mappingCount());
        queueSend(userId);
    }

    @OnClose
    public void onClose() {
        websocketSessionMap.remove(this.userId);
        log.info("【用户断开：{}】,【目前总数：{}】", this.userId, websocketSessionMap.mappingCount());
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        if (WebsocketConstant.HEART_BEAT_SIGNAL.equals(message)) {
            log.debug("【心跳：{}】,【来自：{}】", message, this.userId);
            return;
        }
        // TODO 消息上报接收业务处理，视业务处理
        log.info("【消息：{}】,【来自：{}】", message, this.userId);
        sendMessage(this.userId, message);
    }

    @OnError
    public void onError(Session session, Throwable throwable) {
        websocketSessionMap.remove(this.userId);
        log.info("【用户异常：{}】,【异常：{}】", this.userId, throwable);
    }

    /**
     * 全量消息
     *
     * @param message
     */
    public void sendMessage(String message) {
        // 全量推送不处理离线，默认推送当前所有在线
        log.info("【全量消息开始推送：{}】", message);
        websocketSessionMap.values().forEach(clientSession -> clientSession.session.getAsyncRemote().sendText(message));
        log.info("【全量消息推送完成：" + message + "】");
    }

    /**
     * 消息直接推送
     *
     * @param userId
     * @param message
     */
    public void sendMessage(String userId, String message) {
        log.info("【单人消息开始推送：{}】,【目标：{}】", message, userId);
        RedisUtil.listOps().llen(WebsocketConstant.persistentQueue(userId));
        Optional.ofNullable(userId).map(websocketSessionMap::get).ifPresent(clientSession -> {
            clientSession.session.getAsyncRemote().sendText(message);
            log.info("【单人消息推送完成：{}】,【目标：{}】", message, userId);
        });
    }

    /**
     * 离线持久化消息推送
     *
     * @param userId
     */
    public void queueMessage(String userId) {
        log.info("【持久化消息开始推送】,【目标：{}】", userId);
        RedisUtil.listOps().llen(WebsocketConstant.persistentQueue(userId));
        Optional.ofNullable(userId).map(websocketSessionMap::get).ifPresent(clientSession -> {
            clientSession.queueSend(userId);
            log.info("【持久化消息推送完成】,【目标：{}】", userId);
        });
    }

    /**
     * 若存在历史消息，优先推送历史消息再做实际推送
     * 不处理推送异常、结果，发起推送动作及认为动作完成
     *
     * @param userId
     */
    @SneakyThrows
    @SuppressWarnings({"SynchronizeOnNonFinalField"})
    private void queueSend(String userId) {
        long msgCount = Optional.ofNullable(RedisUtil.listOps().llen(WebsocketConstant.persistentQueue(userId))).map(count -> NumberUtil.max(count, 0L)).orElse(0L);
        while (msgCount-- > 0) {
            String recentMessage = RedisUtil.listOps().lpop(WebsocketConstant.persistentQueue(userId));
            if (recentMessage == null) {
                break;
            }
            // 单个websocket session通道需要串行收发避免socket占用
            synchronized (session) {
                session.getBasicRemote().sendText(recentMessage);
            }
        }
    }
}
